/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * ---------------------------------------------------------------------------------------
 *
 * imcs_flush_insert.cpp
 *
 * IDENTIFICATION
 * src/gausskernel/storage/heapam/imcstore/imcs_flush_insert.cpp
 *
 * ---------------------------------------------------------------------------------------
 */
#include "postgres.h"
#include "knl/knl_variable.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "access/genam.h"
#include "access/cstore_delta.h"
#include "access/reloptions.h"
#include "catalog/catalog.h"
#include "utils/aiomem.h"
#include "utils/datum.h"
#include "utils/gs_bitmap.h"
#include "utils/fmgroids.h"
#include "utils/snapmgr.h"
#include "utils/relcache.h"
#include "catalog/pg_type.h"
#include "access/imcs/imcs_am.h"
#include "access/imcs/imcu_storage.h"
#include "utils/builtins.h"
#include "executor/executor.h"
#include "replication/dataqueue.h"
#include "storage/lmgr.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "pgxc/pgxc.h"
#include "access/heapam.h"
#include "utils/date.h"
#include "access/imcs/imcs_allocspace.h"
#include "storage/ipc.h"
#include "catalog/pg_partition_fn.h"
#include "libpq/pqformat.h"
#include "commands/tablespace.h"
#include "optimizer/var.h"
#include "catalog/index.h"
#include "storage/time_series_compress.h"
#include "access/imcs/imcs_flush_insert.h"
#include "access/imcs/imcs.h"
#include "utils/dynahash.h"

#define MAX_CU_WRITE_REQSIZ 64

#define cstore_backwrite_quantity 8192

#define ENABLE_DELTA(batch)                                                               \
    ((batch) != NULL && ((g_instance.attr.attr_storage.enable_delta_store && is_end()) && \
        ((batch)->m_rows_curnum < m_delta_rows_threshold)))

extern void CuListWriteAbort(int code, Datum arg);

static inline void CuAppendNullValue(int which, void *cu_ptr)
{
    ((IMCU *)cu_ptr)->AppendNullValue(which);
}

static inline int str_to_uint64(const char *str, uint64 *val)
{
    char *end = NULL;
    uint64 value = 0;

    Assert(str != NULL && val != NULL);
    uint32 str_len = strlen(str);
    if (str_len == 0) {
        return -1;
    }

    /* clear errno before convert */
    errno = 0;
#ifdef WIN32
    value = _strtoui64(str, &end, 10);
#else
    value = strtoul(str, &end, 10);
#endif
    if ((errno != 0) || (end != (str + str_len))) {
        return -1;
    }

    *val = value;
    return 0;
}

/* index of m_form_cu_func_array[] calls */
#define FORMCU_IDX_NONE_NULL 0
#define FORMCU_IDX_HAVE_NULL 1

IMCSFlushInsert::IMCSFlushInsert(_in_ Relation relation, _in_ ResultRelInfo *result_rel_info, _in_ bool is_update_cu,
    _in_ Plan *plan, _in_ ImcsMemInfoArg *argmem_info, TupleDesc tupdesc_with_ctid)
    : m_full_cu_size(RelationGetMaxBatchRows(relation)), m_delta_rows_threshold(RelationGetDeltaRowsThreshold(relation))
{
    m_slot_id = -1;
    m_imi_relation = relation;
    td_tid = tupdesc_with_ctid;

    m_imi_cstor_insert_mem = NULL;
    m_imi_cur_cuid = 0;

    /* set the cstore Insert mem info. */
    InitInsertMemArg(plan, argmem_info);

    int att_no = td_tid->natts;

    m_cu_storage = (IMCUStorage **)palloc0(sizeof(IMCUStorage *) * att_no);
    m_cu_compress_options = (compression_options *)palloc0(sizeof(compression_options) * att_no);

    for (int i = 0; i < att_no; ++i) {
        if (td_tid->attrs[i].attisdropped) {
            m_cu_storage[i] = NULL;
        } else {
            /* Here we must use physical column id */
            CFileNode cfile_node(m_imi_relation->rd_node, td_tid->attrs[i].attnum, MAIN_FORKNUM);
            m_cu_storage[i] = New(CurrentMemoryContext)IMCUStorage(cfile_node);
        }
        /* init compression filter */
        m_cu_compress_options[i].reset();
    }

    /* set the compression level and extra modes. */
    m_compress_modes = 0;
    heaprel_set_compressing_modes(m_imi_relation, &m_compress_modes);

    /* set update flag */
    m_is_update = is_update_cu;

    BeginBatchInsert();
}

IMCSFlushInsert::~IMCSFlushInsert()
{
    m_imi_relation = NULL;
    m_set_min_max_funcs = NULL;
    m_cu_compress_options = NULL;
    m_cu_desc_pptr = NULL;
    m_cu_storage = NULL;
    m_cu_pptr = NULL;
    m_imi_cstor_insert_mem = NULL;
}

inline void IMCSFlushInsert::FlushInsertSetFormCuFuncArray(Form_pg_attribute attr, int col)
{
    if (ATT_IS_NUMERIC_TYPE(attr->atttypid) && attr->atttypmod != -1) {
        /* Numeric data type. */
        m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_NONE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCutNumeric<false>;
        m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_HAVE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCutNumeric<true>;
    } else if (ATT_IS_CHAR_TYPE(attr->atttypid)) {
        /* String data type. */
        m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_NONE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCutNumString<false>;
        m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_HAVE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCutNumString<true>;
    } else {
        /* copy data directly for the other data types */
        m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_NONE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCut<false>;
        m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_HAVE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCut<true>;
    }
}

void IMCSFlushInsert::Destroy()
{
    /* Step 2: Destroy all the new objects. */
    int att_no = td_tid->natts;
    for (int col = 0; col < att_no; ++col) {
        if (m_cu_desc_pptr[col]) {
            delete m_cu_desc_pptr[col];
            m_cu_desc_pptr[col] = NULL;
        }
        if (m_cu_storage[col])
            DELETE_EX(m_cu_storage[col]);
    }

#ifdef USE_ASSERT_CHECKING
    if (m_cu_pptr) {
        BFIO_RUN()
        {
            for (int i = 0; i < att_no; i++) {
                Assert(m_cu_pptr[i] == NULL);
            }
        }
        BFIO_END();
    }
#endif

    /* Step 4: Reset all the pointers */
    m_form_cu_func_array = NULL;
    m_set_min_max_funcs = NULL;
    m_cu_storage = NULL;
    m_cu_desc_pptr = NULL;
    m_cu_pptr = NULL;
    m_cu_compress_options = NULL;

    if (m_imi_cstor_insert_mem) {
        pfree_ext(m_imi_cstor_insert_mem);
    }
}

void IMCSFlushInsert::InitInsertMemArg(Plan *plan, ImcsMemInfoArg *argmem_info)
{
    bool has_pck = false;
    int partial_cluster_rows = 0;

    /* get cluster key */
    partial_cluster_rows = RelationGetPartialClusterRows(m_imi_relation);
    if (m_imi_relation->rd_rel->relhasclusterkey) {
        has_pck = true;
    }

    m_imi_cstor_insert_mem = (ImcsMemInfoArg *)palloc0(sizeof(struct ImcsMemInfoArg));
    if (plan != NULL && plan->operatorMemKB[0] > 0) {
        if (!has_pck) {
            m_imi_cstor_insert_mem->MemInsert = plan->operatorMemKB[0];
            m_imi_cstor_insert_mem->MemSort = 0;
        } else {
            m_imi_cstor_insert_mem->MemInsert = (int)((double)plan->operatorMemKB[0] * (double)(m_full_cu_size * 3) /
                (double)(m_full_cu_size * 3 + partial_cluster_rows));
            m_imi_cstor_insert_mem->MemSort = plan->operatorMemKB[0] - m_imi_cstor_insert_mem->MemInsert;
        }
        m_imi_cstor_insert_mem->canSpreadmaxMem = plan->operatorMaxMem;
        m_imi_cstor_insert_mem->spreadNum = 0;
        m_imi_cstor_insert_mem->partitionNum = 1;
        MEMCTL_LOG(DEBUG2,
            "IMCSFlushInsert(init plan):Insert workmem is : %dKB, sort workmem: %dKB,can spread "
            "mem is %dKB.",
            m_imi_cstor_insert_mem->MemInsert, m_imi_cstor_insert_mem->MemSort,
            m_imi_cstor_insert_mem->canSpreadmaxMem);
    } else if (argmem_info != NULL) {
        Assert(argmem_info->partitionNum > 0);
        m_imi_cstor_insert_mem->canSpreadmaxMem = argmem_info->canSpreadmaxMem;
        m_imi_cstor_insert_mem->MemInsert = argmem_info->MemInsert;
        m_imi_cstor_insert_mem->MemSort = argmem_info->MemSort / argmem_info->partitionNum;
        if (m_imi_cstor_insert_mem->MemSort < SORT_MIM_MEM)
            m_imi_cstor_insert_mem->MemSort = SORT_MIM_MEM;
        m_imi_cstor_insert_mem->spreadNum = argmem_info->spreadNum;
        m_imi_cstor_insert_mem->partitionNum = argmem_info->partitionNum;
        MEMCTL_LOG(DEBUG2,
            "IMCSFlushInsert(init argmem_info):Insert workmem is : %dKB, one paritition sort workmem: %dKB,"
            "partition totalnum is %d, can spread mem is %dKB.",
            m_imi_cstor_insert_mem->MemInsert, m_imi_cstor_insert_mem->MemSort, m_imi_cstor_insert_mem->partitionNum,
            m_imi_cstor_insert_mem->canSpreadmaxMem);
    } else {
        m_imi_cstor_insert_mem->canSpreadmaxMem = 0;
        m_imi_cstor_insert_mem->MemInsert = u_sess->attr.attr_storage.partition_max_cache_size;
        m_imi_cstor_insert_mem->MemSort = u_sess->attr.attr_storage.psort_work_mem;
        m_imi_cstor_insert_mem->spreadNum = 0;
        m_imi_cstor_insert_mem->partitionNum = 1;
    }
}
void IMCSFlushInsert::FlushInsertInitFuncPtr()
{
    int att_no = td_tid->natts;
    FormData_pg_attribute *attrs = td_tid->attrs;

    m_set_min_max_funcs = (ImcsFuncSetMinMax *)palloc0(att_no * sizeof(ImcsFuncSetMinMax));
    m_form_cu_func_array = (FormCUFuncArray *)palloc0(sizeof(FormCUFuncArray) * att_no);
    m_cu_desc_pptr = (IMCUDesc **)palloc0(att_no * sizeof(IMCUDesc *));

    for (int col = 0; col < att_no; ++col) {
        if (!attrs[col].attisdropped) {
            m_set_min_max_funcs[col] = ImcsGetMinMaxFunc(attrs[col].atttypid);
            FlushInsertSetFormCuFuncArray(&attrs[col], col);
            m_cu_desc_pptr[col] = New(CurrentMemoryContext)IMCUDesc;
        } else {
            m_set_min_max_funcs[col] = NULL;
            m_cu_desc_pptr[col] = NULL;
            m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_NONE_NULL] = NULL;
            m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_HAVE_NULL] = NULL;
        }
    }
}

void IMCSFlushInsert::FlushInsertInitColSpaceAlloc()
{
    int att_no = td_tid->natts;
    FormData_pg_attribute *attrs = td_tid->attrs;

    AttrNumber *attr_ids = (AttrNumber *)palloc0(sizeof(AttrNumber) * att_no);
    for (int i = 0; i < att_no; ++i) {
        attr_ids[i] = attrs[i].attnum;

        /* Set all column use APPEND_ONLY */
        if (!attrs[i].attisdropped)
            m_cu_storage[i]->SetAllocateStrategy(APPEND_ONLY);
    }
    pfree_ext(attr_ids);
}

void IMCSFlushInsert::BeginBatchInsert()
{
    FlushInsertInitFuncPtr();

    FlushInsertInitColSpaceAlloc();

    m_cu_pptr = (IMCU **)palloc0(sizeof(IMCU *) * td_tid->natts);
}

void IMCSFlushInsert::BatchInsertCommon(IMCSBulkloadRows *batch_row_ptr, IMCU **old_cus, IMCUDesc **old_cus_descs,
    bool multithread, IMCSDesc *imcs_desc)
{
    if (unlikely(batch_row_ptr == NULL || batch_row_ptr->m_rows_curnum == 0))
        return;

    int col = 0;
    bool batch_appended = false;
    bool use_exist_imcu_cache_desc = false;

    CHECK_FOR_INTERRUPTS();
    /* step 1: form IMCU and IMCUDesc; */
    for (col = 0; col < batch_row_ptr->m_attr_num; ++col) {
        if (!td_tid->attrs[col].attisdropped) {
            m_cu_compress_options[col].m_sampling_finished = true;
            m_cu_pptr[col] =
                FlushInsertFormCu(col, batch_row_ptr, m_cu_desc_pptr[col], td_tid, old_cus_descs == nullptr);
        }
    }
    if (m_is_update)
        pgstat_count_cu_update(m_imi_relation, batch_row_ptr->m_rows_curnum);
    else
        pgstat_count_cu_insert(m_imi_relation, batch_row_ptr->m_rows_curnum);


    uint32 max_cu_id = 0;
    IMCU_CACHE->AcquireImcsDescLock(RelationGetRelid(m_imi_relation), LW_EXCLUSIVE);

    if (multithread)
        m_imi_cur_cuid = imcs_desc->rowgroup_num;
    for (col = 0; col < batch_row_ptr->m_attr_num; ++col) {
        if (td_tid->attrs[col].attisdropped)
            continue;

        IMCUDesc *cu_desc = m_cu_desc_pptr[col];

        int imcs_key = td_tid->attrs[col].attnum;
        if (imcs_key == VirtualCtidColID) {
            imcs_key = 0;
        }

        cu_desc->cu_id = m_imi_cur_cuid;

        if (likely(cu_desc->cu_size > 0)) {
            cu_desc->cu_pointer = (uint64)0;
        }

        use_exist_imcu_cache_desc = false;
        if (old_cus_descs) {
            int imcs_col = (col == (batch_row_ptr->m_attr_num - 1)) ? 0 : (col + 1);
            AppendImcus(old_cus[imcs_col], old_cus_descs[imcs_col], m_cu_pptr[col], cu_desc,
                use_exist_imcu_cache_desc, col);
            batch_appended = true;
        }

        if (!use_exist_imcu_cache_desc) {
            bool has_found = false;
            DataSlotTag data_slot_tag = IMCU_CACHE->InitCUSlotTag((RelFileNodeOld *)&m_imi_relation->rd_node, imcs_key,
                cu_desc->cu_id, cu_desc->cu_pointer);
            if (!cu_desc->IsNullCU()) {
                m_slot_id = IMCU_CACHE->ReserveImcuBlock(&data_slot_tag, cu_desc, has_found, m_cu_pptr[col]);
            }

            int attlen = td_tid->attrs[col].attlen;
            max_cu_id = IMCU_CACHE->SaveImcuDesc(RelationGetRelid(m_imi_relation), cu_desc, imcs_key, attlen);
        }

        if (m_cu_pptr[col]) {
            m_cu_pptr[col] = NULL;
        }
    }

    // increase row group id
    if (!batch_appended) {
        IMCU_CACHE->IncreaseImcsRowGroupNum(RelationGetRelid(m_imi_relation), max_cu_id);
        m_imi_cur_cuid++;
    }
    IMCU_CACHE->ReleaseImcsDescLock(RelationGetRelid(m_imi_relation));
}

IMCU *IMCSFlushInsert::FlushInsertFormCu(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cu_desc_ptr,
    TupleDesc tupdesc_with_ctid, bool compress = true)
{
    FormData_pg_attribute *attrs = tupdesc_with_ctid->attrs;
    int attlen = attrs[col].attlen;
    IMCU *cu_ptr = NULL;

    cu_ptr = New(CurrentMemoryContext)IMCU(attlen, attrs[col].atttypmod, attrs[col].atttypid);
    cu_desc_ptr->Reset();

    int fun_idx = batch_row_ptr->m_vectors[col].m_values_nulls.m_has_null ? FORMCU_IDX_HAVE_NULL : FORMCU_IDX_NONE_NULL;
    (this->*(m_form_cu_func_array[col].col_form_cu[fun_idx]))(col, batch_row_ptr, cu_desc_ptr, cu_ptr);

    cu_desc_ptr->magic = GetCurrentTransactionIdIfAny();
    if (!(cu_desc_ptr->IsNullCU())) { // todo  how calculate the cu size   && !(cu_desc_ptr->is_same_val_cu())

        m_cu_temp_info.m_valid_minmax = !NeedToRecomputeMinMax(attrs[col].atttypid);
        if (m_cu_temp_info.m_valid_minmax) {
            m_cu_temp_info.m_min_value = ConvertToInt64Data(cu_desc_ptr->cu_min, attlen);
            m_cu_temp_info.m_max_value = ConvertToInt64Data(cu_desc_ptr->cu_max, attlen);
        }
        m_cu_temp_info.m_options = (m_cu_compress_options + col);
        cu_ptr->m_tmpinfo = &m_cu_temp_info;

        /* Magic number is for checking IMCU data */
        cu_ptr->SetMagic(cu_desc_ptr->magic);
        if (compress) {
            cu_ptr->Compress(batch_row_ptr->m_rows_curnum, m_compress_modes, ALIGNOF_CUSIZE);
            cu_desc_ptr->cu_size = cu_ptr->GetCUSize();
        }
    }
    cu_desc_ptr->row_count = batch_row_ptr->m_rows_curnum;
    cu_desc_ptr->flag = IMCUDescInMemory;

    return cu_ptr;
}

bool IMCSFlushInsert::FlushInsertTryEncodeNumeric(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cu_desc_ptr,
    IMCU *cu_ptr, bool has_null)
{
    char *ptr = NULL;
    int rows = batch_row_ptr->m_rows_curnum;
    int not_exact_size = 0;
    int exact_size = 0;
    int rc = 0;

    if (COMPRESS_NO == heaprel_get_compression_from_modes(m_compress_modes)) {
        return false;
    }

    numerics_statistics_out phase1_out;
    rc = memset_s(&phase1_out, sizeof(numerics_statistics_out), 0, sizeof(numerics_statistics_out));
    securec_check(rc, "\0", "\0");
    phase1_out.success = (bool *)palloc0(sizeof(bool) * rows);
    phase1_out.ascale_codes = (char *)palloc0(sizeof(char) * rows);
    phase1_out.int64_values = (int64 *)palloc0(sizeof(int64) * rows + 8);
    phase1_out.int32_values = (int32 *)palloc0(sizeof(int32) * rows + 8);

    numerics_cmprs_out phase2_out;
    rc = memset_s(&phase2_out, sizeof(numerics_cmprs_out), 0, sizeof(numerics_cmprs_out));
    securec_check(rc, "\0", "\0");

    phase2_out.filter = m_cu_compress_options + col;

    BatchNumeric batch = { batch_row_ptr->m_vectors[col].m_values_nulls.m_vals_points,
                           batch_row_ptr->m_vectors[col].m_values_nulls.m_null_bitmap,
                           batch_row_ptr->m_rows_curnum,
                           has_null,
                           CuAppendNullValue,
                           (void *)cu_ptr };
    if (NumericCompressBatchValues(&batch, &phase1_out, &phase2_out, &not_exact_size, ALIGNOF_CUSIZE)) {
        Assert(cu_ptr->m_srcDataSize == 0);
        if (unlikely(cu_ptr->m_srcData + not_exact_size > cu_ptr->m_srcBuf + cu_ptr->m_srcBufSize)) {
            Assert(not_exact_size >= 0);
            cu_ptr->ReallocMem((Size)cu_ptr->m_srcBufSize + (uint32)not_exact_size);
        }

        ptr = NumericCopyCompressedBatchValues(cu_ptr->m_srcData, &phase1_out, &phase2_out);

        if (!m_cu_compress_options[col].m_sampling_finished) {
            m_cu_compress_options[col].set_numeric_flags(*(uint16 *)cu_ptr->m_srcData);
        }

        exact_size = (ptr - cu_ptr->m_srcData);

        cu_ptr->m_srcDataSize = exact_size;

        (void)IMCStore::SetCudescModeForMinMaxVal(false, false, has_null, 0, -1, cu_desc_ptr);

        NumericCompressReleaseResource(&phase1_out, &phase2_out);

        cu_ptr->m_infoMode |= (CU_IntLikeCompressed | CU_DSCALE_NUMERIC);

        return true;
    }

    NumericCompressReleaseResource(&phase1_out, &phase2_out);
    return false;
}

template <bool bpchar_type, bool has_null, bool tem_has_min_max_func>
bool IMCSFlushInsert::FlushInsertFormNumberStringCu(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cu_desc_ptr,
    IMCU *cu_ptr)
{
    int attlen = this->td_tid->attrs[col].attlen;
    int rows = batch_row_ptr->m_rows_curnum;
    imcs_bulkload_datums *batch_values = &(batch_row_ptr->m_vectors[col].m_values_nulls);
    uint64 data = 0;
    char *data_buf = (char *)palloc0(sizeof(char) * (MAX_LEN_CHAR_TO_BIGINT_BUF) * (rows));
    uint32 data_count = 0;
    uint32 idx = 0;
    uint32 next_data_pos = 0;
    uint32 max_str_len = 0;
    uint32 bpchar_data_len = 0;
    /* the total length excludes the var-header size of all the values */
    uint32 total_len = 0;
    ImcsFuncSetMinMax *min_max_func = this->m_set_min_max_funcs + col;
    bool first_flag = true;
    bool has_min_max_func = false;

    for (int i = 0; i < rows; ++i) {
        if (has_null && batch_values->is_null(i)) {
            cu_ptr->AppendNullValue(i);
            continue;
        }

        Datum v = batch_values->get_datum(i);
        char *p = DatumGetPointer(v);
        Assert(p != NULL);
        uint32 len = VARSIZE_ANY_EXHDR(p);

        if (unlikely(len == 0)) {
            pfree(data_buf);
            return false;
        }
        Assert(len > 0);

        total_len += len;

        if (bpchar_type) {
            bpchar_data_len = len;
        }
        /* dymic length char type, check length first */
        if (!bpchar_type && len > MAX_LEN_CHAR_TO_BIGINT) {
            pfree(data_buf);
            return false;
        }
        if (max_str_len < len + VARHDRSZ) {
            max_str_len = len + VARHDRSZ;
        }
        char *ptr = VARDATA_ANY(p);
        /* first byte must be 1~9 */
        if (*ptr > '9' || *ptr < '1') {
            pfree(data_buf);
            return false;
        }

        /* this index used to write cannot be overflow. */
        Assert(idx < MAX_LEN_CHAR_TO_BIGINT_BUF * (uint32)rows);

        /* data_buf store valid data num */
        data_buf[idx] = *ptr;
        --len;
        ++ptr;
        ++idx;
        while (len > 0) {
            /* next byte must be 0~9 or ' ' */
            if (*ptr > '9' || *ptr < '0') {
                if (bpchar_type) {
                    /* fix length char, need check ' ' after num */
                    break;
                } else {
                    /* dymic length char can return not ok */
                    pfree(data_buf);
                    return false;
                }
            }

            /* fix length char, need check length and ' ' which all the buf left.
             * we will write extra one byte next step, so we have to check
             * the current length of digits. because not all 20 digits can be
             * converted to uint64 value, we have to finish if this case happens.
             */
            if (bpchar_type && bpchar_data_len - len >= MAX_LEN_CHAR_TO_BIGINT) {
                pfree(data_buf);
                return false;
            }

            data_buf[idx] = *ptr;
            --len;
            ++ptr;
            ++idx;
        }

        /* this index used to write cannot be overflow. */
        Assert(idx < MAX_LEN_CHAR_TO_BIGINT_BUF * (uint32)rows);
        if (bpchar_type) {
            /* each integer digits cannot be greater than 19 */
            Assert(bpchar_data_len - len <= MAX_LEN_CHAR_TO_BIGINT);
            MEMCTL_LOG(DEBUG2, "Assert each integer digits cannot be greater than 19 when bpchar_type is true");
        }

        /* set complete flag for function strtoul need it */
        data_buf[idx] = '\0';
        if (bpchar_type) {
            while (len > 0) {
                if (*ptr != ' ') {
                    pfree(data_buf);
                    return false;
                }
                --len;
                ++ptr;
            }
        }

        /* make right idx for store next data value */
        ++data_count;
        next_data_pos += MAX_LEN_CHAR_TO_BIGINT_BUF;
        idx = next_data_pos;

        if (tem_has_min_max_func) {
            (*(min_max_func))(v, cu_desc_ptr, &first_flag);
            has_min_max_func = true;
        }
        first_flag = false;
    }

    /* for deform, store count */
    cu_ptr->AppendValue(data_count, sizeof(uint64));

    /* do convert, must sucess */
    char *tmp_data_buf = data_buf;
    for (uint32 i = 0; i < data_count; ++i) {
        int ret = str_to_uint64(tmp_data_buf, &data);
        if (ret != 0) {
            pfree(data_buf);
            return false;
        }
        /* append each integer value */
        cu_ptr->AppendValue(data, sizeof(uint64));
        tmp_data_buf += MAX_LEN_CHAR_TO_BIGINT_BUF;
    }

    /* for deform, store src data size but excludes
     * their var-header size.
     */
    cu_ptr->AppendValue(total_len, sizeof(uint64));

    (void)IMCStore::SetCudescModeForMinMaxVal(first_flag, has_min_max_func, has_null, max_str_len, attlen, cu_desc_ptr);

    pfree(data_buf);
    cu_ptr->m_infoMode |= CU_IntLikeCompressed;
    return true;
}

template <bool has_null>
bool IMCSFlushInsert::FlushInsertTryFormNumberStringCu(int col, IMCSBulkloadRows *batch_row_ptr,
    IMCUDesc *cu_desc_ptr, IMCU *cu_ptr, uint32 atttypid)
{
    bool ret = false;
    ImcsFuncSetMinMax func = *(this->m_set_min_max_funcs + col);

    if (COMPRESS_NO == heaprel_get_compression_from_modes(this->m_compress_modes)) {
        return ret;
    }

    if (atttypid == BPCHAROID) {
        /* for type define bpchar, we do not know the length, so we don't do the change */
        if (this->td_tid->attrs[col].atttypmod == -1) {
            return ret;
        }
        if (func == NULL) {
            ret = this->FlushInsertFormNumberStringCu<true, has_null, false>(col, batch_row_ptr, cu_desc_ptr, cu_ptr);
        } else {
            ret = this->FlushInsertFormNumberStringCu<true, has_null, true>(col, batch_row_ptr, cu_desc_ptr, cu_ptr);
        }
    } else {
        if (func == NULL) {
            ret = this->FlushInsertFormNumberStringCu<false, has_null, false>(col, batch_row_ptr, cu_desc_ptr, cu_ptr);
        } else {
            ret = this->FlushInsertFormNumberStringCu<false, has_null, true>(col, batch_row_ptr, cu_desc_ptr, cu_ptr);
        }
    }

    return ret;
}

Size IMCSFlushInsert::FlushInsertFormCutInitMem(IMCU *cu_ptr, IMCSBulkloadRows *batch_row_ptr, int col, bool has_null)
{
    imcs_bulkload_vector *vector = batch_row_ptr->m_vectors + col;

    /* compute the total size */
    Size dt_size = vector->data_size();
    Size initial_size = cu_ptr->GetCUHeaderSize() + sizeof(Datum) + dt_size;
    initial_size = MAXALIGN(initial_size);
    cu_ptr->InitMem((uint32)initial_size, batch_row_ptr->m_rows_curnum, has_null);
    return dt_size;
}

void IMCSFlushInsert::FlushInsertFormCutCopyMem(IMCU *cu_ptr, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cu_desc_ptr,
    Size dt_size, int col, bool has_null)
{
    imcs_bulkload_vector *vector = batch_row_ptr->m_vectors + col;
    Form_pg_attribute attr = &this->td_tid->attrs[col];

    /* copy null-bitmap */
    if (has_null) {
        /* we have computed the total memory size including null bitmap */
        Size bpsize = bitmap_size(batch_row_ptr->m_rows_curnum);
        errno_t rc = memcpy_s(cu_ptr->m_nulls, bpsize, vector->m_values_nulls.m_null_bitmap, bpsize);
        securec_check(rc, "\0", "\0");
    }

    /* copy all the data */
    vector->data_copy(cu_ptr->m_srcData);
    cu_ptr->m_srcDataSize = dt_size;

    /* set min/max for IMCU Desc */
    bool has_min_max_func = !ImcsIsCompareDatumDummyFunc(vector->m_minmax.m_compare);
    int max_str_len = vector->m_minmax.m_varstr_maxlen;
    vector->m_minmax.m_finish_compare(vector->m_minmax.m_min_buf, vector->m_minmax.m_max_buf, cu_desc_ptr);

    /* Check whether IMCU is filled with the same value
     * Check whether IMCU is filled with all NULL value
     */
    (void)IMCStore::SetCudescModeForMinMaxVal(vector->m_values_nulls.m_all_null, has_min_max_func, has_null,
        max_str_len, attr->attlen, cu_desc_ptr);
}

template <bool has_null>
void IMCSFlushInsert::FlushInsertFormCut(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cu_desc_ptr,
    IMCU *cu_ptr)
{
    Size dt_size = this->FlushInsertFormCutInitMem(cu_ptr, batch_row_ptr, col, has_null);
    this->FlushInsertFormCutCopyMem(cu_ptr, batch_row_ptr, cu_desc_ptr, dt_size, col, has_null);
}

template <bool has_null>
void IMCSFlushInsert::FlushInsertFormCutNumeric(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cu_desc_ptr,
    IMCU *cu_ptr)
{
    Size dt_size = this->FlushInsertFormCutInitMem(cu_ptr, batch_row_ptr, col, has_null);

    if (this->FlushInsertTryEncodeNumeric(col, batch_row_ptr, cu_desc_ptr, cu_ptr, has_null)) {
        return;
    }

    /* clean-up work here.
     * ignore the nulls memory even though it will be set later again.
     */
    cu_ptr->m_srcDataSize = 0;

    this->FlushInsertFormCutCopyMem(cu_ptr, batch_row_ptr, cu_desc_ptr, dt_size, col, has_null);

    /* change m_form_cu_func_array[col] to FormCUTCommon()
     * if the first time of encoding-numeric fails
     */
    if (!this->m_cu_compress_options[col].m_sampling_finished) {
        this->m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_NONE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCut<false>;
        this->m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_HAVE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCut<true>;
    }
}

template <bool has_null>
void IMCSFlushInsert::FlushInsertFormCutNumString(int col, IMCSBulkloadRows *batch_row_ptr, IMCUDesc *cu_desc_ptr,
    IMCU *cu_ptr)
{
    Form_pg_attribute attr = &this->td_tid->attrs[col];

    Size data_size = this->FlushInsertFormCutInitMem(cu_ptr, batch_row_ptr, col, has_null);

    if (this->FlushInsertTryFormNumberStringCu<has_null>(col, batch_row_ptr, cu_desc_ptr, cu_ptr, attr->atttypid)) {
        return;
    }

    /* clean-up work here. Ignore the nulls memory even though it will be set later again. */
    cu_ptr->m_srcDataSize = 0;

    this->FlushInsertFormCutCopyMem(cu_ptr, batch_row_ptr, cu_desc_ptr, data_size, col, has_null);
    /* change m_form_cu_func_array[col] to FormCUTCommon()	if the first time of encoding-number-string fails */
    if (!this->m_cu_compress_options[col].m_sampling_finished) {
        this->m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_NONE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCut<false>;
        this->m_form_cu_func_array[col].col_form_cu[FORMCU_IDX_HAVE_NULL] =
            &IMCSFlushInsert::FlushInsertFormCut<true>;
    }
}

void IMCSFlushInsert::UpdateAppendedImcu(IMCU *dest_cu, IMCU *src_cu, IMCUDesc *dest_cu_desc, IMCUDesc *src_cu_desc,
    bool is_old_desc_null_cu)
{
    int tot_row_cnt = dest_cu_desc->row_count + src_cu_desc->row_count;
    UpdateAppendedImcuBuf(dest_cu, src_cu, dest_cu_desc, src_cu_desc, is_old_desc_null_cu);

    dest_cu->m_nulls = dest_cu->HasNullValue() ? (unsigned char *)dest_cu->m_srcBuf : nullptr;
    int null_size = dest_cu->HasNullValue() ? bitmap_size(tot_row_cnt) : 0;
    dest_cu->m_srcData = dest_cu->HasNullValue() ? &dest_cu->m_srcBuf[null_size] : dest_cu->m_srcBuf;
    dest_cu->m_srcDataSize += src_cu ? src_cu->m_srcDataSize : 0;
    dest_cu->m_bpNullRawSize = null_size;
    dest_cu->m_infoMode |= (null_size > 0 ? CU_HasNULL : 0);

    if (dest_cu->m_offset) {
        CStoreMemAlloc::Pfree(dest_cu->m_offset, !dest_cu->m_inCUCache);
    }
    dest_cu->m_offset = NULL;
    dest_cu->m_offsetSize = 0;
    dest_cu->HasNullValue() ? dest_cu->FormValuesOffset<true>(tot_row_cnt) :
                              dest_cu->FormValuesOffset<false>(tot_row_cnt);
}

void IMCSFlushInsert::UpdateAppendedImcuDesc(IMCUDesc *old_cu_desc, IMCUDesc *new_cu_desc, int col_idx)
{
    IMCUDesc *dest_cu_desc = (old_cu_desc->IsNullCU() ? new_cu_desc : old_cu_desc);
    IMCUDesc *src_cu_desc = (old_cu_desc->IsNullCU() ? old_cu_desc : new_cu_desc);

    dest_cu_desc->cu_id = old_cu_desc->cu_id;
    dest_cu_desc->row_count += src_cu_desc->row_count;

    // set cu_mode
    bool full_nulls = old_cu_desc->IsNullCU() && new_cu_desc->IsNullCU();
    bool has_null = (old_cu_desc->CUHasNull() || new_cu_desc->CUHasNull()) ||
        (old_cu_desc->IsNullCU() && !new_cu_desc->IsNullCU()) || (!old_cu_desc->IsNullCU() && new_cu_desc->IsNullCU());

    int attlen = td_tid->attrs[col_idx].attlen;
    dest_cu_desc->cu_mode = 0;
    ImcsFuncSetMinMax min_max_func = ImcsGetMinMaxFunc(td_tid->attrs[col_idx].atttypid);

    // set min and max value of CU
    if (min_max_func && !src_cu_desc->IsNullCU()) {
        bool first = dest_cu_desc->row_count == 0 ? true : false;
        Datum cu_min;
        Datum cu_max;
        if (attlen > 0 && attlen <= 8) {
            cu_min = *(Datum *)(src_cu_desc->cu_min);
            cu_max = *(Datum *)(src_cu_desc->cu_max);
        } else if (attlen == 12 || attlen == 16) {
            cu_min = PointerGetDatum(src_cu_desc->cu_min);
            cu_max = PointerGetDatum(src_cu_desc->cu_max);
        } else {
            cu_min = PointerGetDatum(src_cu_desc->cu_min + 1);
            Size len_min = (Size)(unsigned char)src_cu_desc->cu_min[0];
            Assert(len_min < MIN_MAX_LEN);

            char tmp_str_min[MIN_MAX_LEN + 4];
            if (attlen == -1) {
                Size varLen = len_min + VARHDRSZ_SHORT;
                SET_VARSIZE_SHORT(tmp_str_min, varLen);
                errno_t rc = memcpy_s(VARDATA_ANY(tmp_str_min), sizeof(tmp_str_min) - VARHDRSZ_SHORT,
                    DatumGetPointer(cu_min), len_min);
                securec_check(rc, "", "");
                cu_min = PointerGetDatum(tmp_str_min);
            }

            cu_max = PointerGetDatum(src_cu_desc->cu_max + 1);
            Size len_max = (Size)(unsigned char)src_cu_desc->cu_max[0];
            Assert(len_max < MIN_MAX_LEN);

            char tmp_str_max[MIN_MAX_LEN + 4];
            if (attlen == -1) {
                Size varLen = len_max + VARHDRSZ_SHORT;
                SET_VARSIZE_SHORT(tmp_str_max, varLen);
                errno_t rc = memcpy_s(VARDATA_ANY(tmp_str_max), sizeof(tmp_str_max) - VARHDRSZ_SHORT,
                    DatumGetPointer(cu_max), len_max);
                securec_check(rc, "", "");
                cu_max = PointerGetDatum(tmp_str_max);
            }
        }

        min_max_func(cu_min, dest_cu_desc, &first);
        min_max_func(cu_max, dest_cu_desc, &first);
    }

    (void)IMCStore::SetCudescModeForMinMaxVal(full_nulls, min_max_func != nullptr, has_null, -1, attlen, dest_cu_desc);
}

void IMCSFlushInsert::UpdateAppendedImcuBuf(IMCU *dest_cu, IMCU *src_cu, IMCUDesc *dest_desc, IMCUDesc *src_desc,
    bool is_old_desc_null_cu)
{
    int src_data_size = src_cu == nullptr ? 0 : src_cu->m_srcDataSize;
    int nulls_size = 0;
    int tot_rows = dest_desc->row_count + src_desc->row_count;

    if (dest_desc->CUHasNull() || dest_desc->IsNullCU() || src_desc->CUHasNull() ||
        (src_desc->IsNullCU() && (src_desc->row_count > 0))) {
        nulls_size = bitmap_size(tot_rows);
    }

    // TODO check the correct size for realloc. The destination null size is also added inside ReallocMem..
    uint32 new_size = src_data_size + dest_cu->m_srcDataSize + nulls_size;
    if (new_size > dest_cu->m_srcBufSize) {
        /* In order to reduce the numeber of realloc we set an extra space. */
        dest_cu->ReallocMem((Size)new_size + 1024 * 1024);
    }

    // copy the new data to the end of the buffer
    errno_t rc;
    if ((size_t)src_data_size > 0) {
        rc = memmove_s(&dest_cu->m_srcBuf[nulls_size + dest_cu->m_srcDataSize], (size_t)src_data_size,
            src_cu->m_srcData, (size_t)src_data_size);
        securec_check(rc, "\0", "\0");
    }
    // move the old data to the new location in the buf
    if (&dest_cu->m_srcBuf[nulls_size] != dest_cu->m_srcData) {
        if ((size_t)dest_cu->m_srcDataSize > 0) {
            rc = memmove_s((void *)&(dest_cu->m_srcBuf[nulls_size]), (size_t)dest_cu->m_srcDataSize,
                (void *)dest_cu->m_srcData, (size_t)dest_cu->m_srcDataSize);
            securec_check(rc, "\0", "\0");
        }
    }
    if (nulls_size == 0)
        return;

    // Fix nulls;
    dest_cu->m_nulls = (unsigned char *)dest_cu->m_srcBuf;
    if (is_old_desc_null_cu) {
        // In case the old cu is null, we need to move the bitmap of the new cu to make space for the old cu bm
        CopyToCuBm(dest_cu, src_desc->row_count, dest_cu, dest_desc->row_count, dest_desc->IsNullCU(),
            dest_desc->CUHasNull());
        // Set the bitmap of the old cu in the new cu
        CopyToCuBm(dest_cu, 0, src_cu, src_desc->row_count, src_desc->IsNullCU(), src_desc->CUHasNull());
    } else {
        // Copy the bitmap of the dest cu (new cu) to the end of the src cu (old cu) bitmap
        CopyToCuBm(dest_cu, dest_desc->row_count, src_cu, src_desc->row_count, src_desc->IsNullCU(),
            src_desc->CUHasNull());
        if (!dest_desc->CUHasNull())
            // The old nulls buf is already copied by realloc if exists thereforee, this update is neede only if this cu
            // has no nulls.
            CopyToCuBm(dest_cu, 0, dest_cu, dest_desc->row_count, dest_desc->IsNullCU(), dest_desc->CUHasNull());
    }
}
void IMCSFlushInsert::CopyToCuBm(IMCU *dest_cu, int offset_in_dest_bm, IMCU *src_cu, int row_cnt_in_src,
    bool src_null_cu, bool src_has_null)
{
    uint32_t mask;
    uint32_t shift;
    int val, new_row_idx, old_row_idx;

    for (old_row_idx = 0, new_row_idx = offset_in_dest_bm; old_row_idx < row_cnt_in_src; old_row_idx++, new_row_idx++) {
        if (src_null_cu) {
            val = 1;
        } else if (src_has_null) {
            shift = (uint32)(old_row_idx % 8);
            mask = (uint32)1 << shift;
            val = ((src_cu->m_srcBuf[old_row_idx / 8] & mask) >> shift);
        } else {
            val = 0;
        }
        if (val) {
            dest_cu->AppendNullValue(new_row_idx);
            dest_cu->m_infoMode |= CU_HasNULL;
        } else
            dest_cu->m_srcBuf[new_row_idx / 8] &= ~(1 << (uint32)(new_row_idx % 8));
    }
}

void IMCSFlushInsert::AppendImcus(IMCU *old_cu, IMCUDesc *old_cu_desc, IMCU *new_cu, IMCUDesc *new_cu_desc,
    bool &use_exist_imcu_cache_desc, int col_idx)
{
    if (old_cu) {
        UpdateAppendedImcu(old_cu, new_cu, old_cu_desc, new_cu_desc, false);
        use_exist_imcu_cache_desc = true;
    } else if (!new_cu_desc->IsNullCU()) {
        // In case the old cu is null cu, the update will be from old to new cu. otherwise the new cu will be appended
        // to the old one
        UpdateAppendedImcu(new_cu, old_cu, new_cu_desc, old_cu_desc, true);
        new_cu->Compress(old_cu_desc->row_count + new_cu_desc->row_count, m_compress_modes, ALIGNOF_CUSIZE);
        new_cu_desc->cu_size = new_cu->GetCUSize();
    }
    UpdateAppendedImcuDesc(old_cu_desc, new_cu_desc, col_idx);
}
